package cn.iotab.iotkit.transport.mqtt.adaptors;

import cn.iotab.iotkit.transport.api.common.message.msg.CommonPayload;
import cn.iotab.iotkit.transport.mqtt.session.MqttSessionCtx;
import cn.iotab.iotkit.transport.mqtt.topic.MqttTopics;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttPublishMessage;
import io.netty.handler.codec.mqtt.MqttPublishVariableHeader;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Optional;

/**
 * @author Andrew Shvayka
 */
@Slf4j
public class JsonMqttAdaptor implements MqttTransportAdaptor {

    protected static final Charset UTF8 = StandardCharsets.UTF_8;

    private static final ByteBufAllocator ALLOCATOR = new UnpooledByteBufAllocator(false);


    @Override
    public CommonPayload convertFromDeviceResponse(String productKey, String deviceId, MqttPublishMessage inbound) throws AdaptorException {
        String payload = validatePayload(productKey, deviceId, inbound.payload(), true);
        try {
            return CommonPayload.of(productKey, deviceId, payload);
        } catch (IllegalStateException ex) {
            throw new AdaptorException(ex);
        }
    }

    @Override
    public CommonPayload convertFromDeviceRequest(String productKey, String deviceId, MqttPublishMessage inbound) throws AdaptorException {
        String payload = validatePayload(productKey, deviceId, inbound.payload(), false);
        try {
            return CommonPayload.of(productKey, deviceId, payload);
        } catch (IllegalStateException ex) {
            throw new AdaptorException(ex);
        }
    }

    @Override
    public Optional<MqttMessage> convertToDeviceRequest(MqttSessionCtx ctx, CommonPayload toDeviceRequest) {
        String topic = MqttTopics.buildDownPostTopic(toDeviceRequest.getProductKey(), toDeviceRequest.getDeviceId());
        return Optional.of(createMqttPublishMsg(ctx, topic, toDeviceRequest.getPayload()));
    }

    @Override
    public Optional<MqttMessage> convertToDeviceResponse(MqttSessionCtx ctx, CommonPayload toDeviceResponse) {
        String topic = MqttTopics.buildDownReplyTopic(toDeviceResponse.getProductKey(), toDeviceResponse.getDeviceId());
        return Optional.of(createMqttPublishMsg(ctx, topic, toDeviceResponse.getPayload()));
    }


    protected MqttPublishMessage createMqttPublishMsg(MqttSessionCtx ctx, String topic, String data) {
        MqttFixedHeader mqttFixedHeader = new MqttFixedHeader(MqttMessageType.PUBLISH, false, ctx.getQoSForTopic(topic), false, 0);
        MqttPublishVariableHeader header = new MqttPublishVariableHeader(topic, ctx.nextMsgId());
        ByteBuf payload = ALLOCATOR.buffer();
        payload.writeBytes(data.getBytes(UTF8));
        return new MqttPublishMessage(mqttFixedHeader, header, payload);
    }


    private static String validatePayload(String productKey, String deviceId, ByteBuf payloadData, boolean isEmptyPayloadAllowed) throws AdaptorException {
        String payload = payloadData.toString(UTF8);
        if (payload == null) {
            log.warn("[{}_{}] Payload is empty!", productKey, deviceId);
            if (!isEmptyPayloadAllowed) {
                throw new AdaptorException(new IllegalArgumentException("Payload is empty!"));
            }
        }
        return payload;
    }
}
